package org.future.task.redis.singleton;

import io.lettuce.core.*;
import io.lettuce.core.api.StatefulRedisConnection;
import io.lettuce.core.api.async.RedisAsyncCommands;
import redis.clients.jedis.HostAndPort;

import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.Executor;
import java.util.function.BiConsumer;

/**
 * @author Dongwei-Chen
 * @Date 2023/12/13 17:13
 * @Description
 */
public class AsyncSingletonManager<K, V> implements AsyncSingletonCommand<K, V> {

    private static StatefulRedisConnection<String, String> connect;

    public AsyncSingletonManager(HostAndPort node, String password) {

        RedisURI.Builder builder = RedisURI.builder().withHost(node.getHost()).withPort(node.getPort()).withDatabase(0);
        if (password != null) {
            builder = builder.withPassword(password);
        }
        RedisURI build = builder.build();

        RedisClient client = RedisClient.create(build);
        connect = client.connect();
        connect.setAutoFlushCommands(false);
    }

    public AsyncSingletonManager(HostAndPort node) {
        this(node, null);
    }

    @Override
    public void batchExecutorAsync(List<K> batch, AsyncSingletonAbstract<K, V> asyncCommandAbstract, Executor executor) {
        RedisAsyncCommands<String, String> async = connect.async();
        CountDownLatch downLatch = new CountDownLatch(batch.size());
        for (K row : batch) {
            try {
                RedisFuture<V> command = asyncCommandAbstract.command(async, row);
                command.whenCompleteAsync((v, throwable) -> {
                    try {
                        asyncCommandAbstract.future(row, v, throwable);
                    } finally {
                        downLatch.countDown();
                    }
                }, executor);
            } catch (Throwable e) {
                downLatch.countDown();
            }
        }
        async.flushCommands();
        try {
            downLatch.await();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    @Override
    public void batchExecutorAsync(List<K> batch, AsyncSingletonAbstract<K, V> asyncCommandAbstract) {
        RedisAsyncCommands<String, String> async = connect.async();
        CountDownLatch downLatch = new CountDownLatch(batch.size());
        for (K row : batch) {
            try {
                RedisFuture<V> command = asyncCommandAbstract.command(async, row);
                command.whenCompleteAsync((v, throwable) -> {
                    try {
                        asyncCommandAbstract.future(row, v, throwable);
                    } finally {
                        downLatch.countDown();
                    }
                });
            } catch (Throwable e) {
                downLatch.countDown();
            }
        }
        async.flushCommands();
        try {
            downLatch.await();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    @Override
    public void batchExecutorSync(List<K> batch, AsyncSingletonAbstract<K, V> asyncCommandAbstract) {
        RedisAsyncCommands<String, String> async = connect.async();
        CountDownLatch downLatch = new CountDownLatch(batch.size());
        for (K row : batch) {
            try {
                RedisFuture<V> command = asyncCommandAbstract.command(async, row);
                command.whenComplete((v, throwable) -> {
                    try {
                        asyncCommandAbstract.future(row, v, throwable);
                    } finally {
                        downLatch.countDown();
                    }
                });
            } catch (Throwable e) {
                downLatch.countDown();
            }
        }
        async.flushCommands();
        try {
            downLatch.await();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    @Override
    public void scanExecutorSync(String cursor, ScanArgs scanArgs) {
        RedisAsyncCommands<String, String> async = connect.async();
        ScanCursor scanCursor = ScanCursor.of(cursor);
        RedisFuture<KeyScanCursor<String>> scan = async.scan(scanCursor, scanArgs);
        scan.whenComplete(new BiConsumer<KeyScanCursor<String>, Throwable>() {
            @Override
            public void accept(KeyScanCursor<String> stringKeyScanCursor, Throwable throwable) {

            }
        });
    }

    @Override
    public void scanExecutorSync(ScanArgs scanArgs) {
        RedisAsyncCommands<String, String> async = connect.async();
        RedisFuture<KeyScanCursor<String>> scan = async.scan(new ScanCursor(), scanArgs);
        scan.whenComplete(new BiConsumer<KeyScanCursor<String>, Throwable>() {
            @Override
            public void accept(KeyScanCursor<String> stringKeyScanCursor, Throwable throwable) {
                List<String> keys = stringKeyScanCursor.getKeys();
                System.out.println(keys.size());
            }
        });
    }
}
